package middleware.p2p.transaction;

import java.net.MalformedURLException;
import java.net.URL;
import java.util.GregorianCalendar;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import javax.ejb.EJB;
import javax.ejb.Stateless;
import javax.jws.WebService;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.xml.ws.Response;

import middleware.p2p.replication.wsclient.CanCommitResponse;
import middleware.p2p.replication.wsclient.DoAbortResponse;
import middleware.p2p.replication.wsclient.DoCommitResponse;
import middleware.p2p.replication.wsclient.ReplicaManager;
import middleware.p2p.replication.wsclient.ReplicaManagerService;
import middleware.p2p.transaction.log.TransactionLoggerLocal;

import org.apache.log4j.Logger;


/**
 * Session Bean implementation class TransactionCoordinator
 */
@Stateless
@WebService(name="TransactionCoordinator", serviceName="TransactionCoordinatorService")
public class TransactionCoordinator implements TransactionCoordinatorRemote {
	@EJB(name="ejb/TransactionRepository",
			beanInterface=middleware.p2p.transaction.TransactionRepositoryLocal.class)
	private TransactionRepositoryLocal repository;
	@EJB(name="ejb/TransactionLogger",
			beanInterface=middleware.p2p.transaction.log.TransactionLoggerLocal.class)
	private TransactionLoggerLocal transLogger;
	private Logger logger=Logger.getLogger(this.getClass());	
    /**
     * Default constructor. 
     */
    public TransactionCoordinator() {
        super();
        try {
	        Context ctx = new InitialContext();
			this.repository = (TransactionRepositoryLocal)ctx.lookup("java:comp/env/ejb/TransactionRepository");
			this.transLogger = (TransactionLoggerLocal)ctx.lookup("java:comp/env/ejb/TransactionLogger");
		} catch (NamingException e) {
			e.printStackTrace();
		}
    }

	@Override
	public boolean abortTransaction(String transId) {
		Transaction trans=this.repository.getTransaction(transId);
		if (trans==null) return false;
		trans.setAborted();
		URL allRM[]= trans.getPartecipants();
		ReplicaManager rmService[]=new ReplicaManager[allRM.length];
		for (int k=0;k<allRM.length;k++) {
			ReplicaManagerService serv=new ReplicaManagerService(allRM[k]);
			ReplicaManager rm=serv.getReplicaManagerPort();
			rmService[k]=rm;
			rm.doAbort(transId);			
		}
		return true;
	}
	@Override
	public TransactionResponse closeTransaction(String transId) {
		long coordinatorTimeouts[]=this.repository.getCoordinatorTimeouts();
		Transaction trans=this.repository.getTransaction(transId);
		if (trans==null) return TransactionResponse.Unknown;
		logger.debug("2PC Algorithm started");
		this.transLogger.writePrepare(transId);
		URL allRM[]= trans.getPartecipants();
		ReplicaManager rmService[]=new ReplicaManager[allRM.length];
		List<Response<CanCommitResponse>> commitVotes=new LinkedList<Response<CanCommitResponse>>();
		for (int k=0;k<allRM.length;k++) {
			ReplicaManagerService serv=new ReplicaManagerService(allRM[k]);
			ReplicaManager rm=serv.getReplicaManagerPort();
			rmService[k]=rm;
			commitVotes.add(rm.canCommitAsync(transId));
			//trans.registerCommitVote(allRM[k], vote);
		}
		long votingSingleTimeout=coordinatorTimeouts[0];
		long votingTotalTimeout=coordinatorTimeouts[1];
		long votingDeltaTimeout=coordinatorTimeouts[2];
		long votingStartTime=GregorianCalendar.getInstance().getTimeInMillis();
		logger.debug("Begin votes collection phase");
		Iterator<Response<CanCommitResponse>> it=commitVotes.iterator();
		int rmIndex=0;
		boolean votingInTime=true;
		while ((it.hasNext())&&votingInTime) {
			Response<CanCommitResponse> voteResp=it.next();
			long currentTime=GregorianCalendar.getInstance().getTimeInMillis();
			if (currentTime-votingStartTime<=votingTotalTimeout-votingDeltaTimeout) {
				try {
					CanCommitResponse vote=voteResp.get(votingSingleTimeout,TimeUnit.MILLISECONDS);
					trans.registerCommitVote(allRM[rmIndex], vote.isReturn());
				} catch (Exception e) {
					logger.error("Voting Single Timeout exceeded",e);
					votingInTime=false;
				}
			} else {
				logger.error("Voting Total Timeout exceeded");
				votingInTime=false;
			}
			rmIndex++;		
		}
		boolean decision=trans.checkVotes()&&votingInTime;
		logger.debug("Voting decision taken. The result is "+(decision?"Commit":"Abort")+".");
		this.transLogger.writeCoordinatorDecision(transId, decision);
		List<Response<DoCommitResponse>> commitAck=new LinkedList<Response<DoCommitResponse>>();
		List<Response<DoAbortResponse>> abortAck=new LinkedList<Response<DoAbortResponse>>();
		for (int k=0;k<allRM.length;k++) {
			ReplicaManager rm=rmService[k];
			if (decision) {
				commitAck.add(rm.doCommitAsync(transId));
			} else {
				abortAck.add(rm.doAbortAsync(transId));							
			}
		}
		long completitionSingleTimeout=coordinatorTimeouts[3];
		long completitionMaxRetry=coordinatorTimeouts[4];
		rmIndex=0;
		logger.debug("Begin acks collecting phase");		
		if (decision) {
			int numRetry=-1;
			while ((commitAck.size()>0)&&(numRetry<completitionMaxRetry)) {
				rmIndex=0;
				Iterator<Response<DoCommitResponse>> ackIt=commitAck.iterator();
				while(ackIt.hasNext()) {
					Response<DoCommitResponse> ack=ackIt.next();
					try {
						ack.get(completitionSingleTimeout, TimeUnit.MILLISECONDS);
						trans.registerCommit(allRM[rmIndex]);
						ackIt.remove();
					} catch (Exception e) {					
						logger.debug("Timeout exceeded: retransmission needed");
					}
					rmIndex++;
				}
				numRetry++;
			}
		} else {
			int numRetry=-1;
			while ((abortAck.size()>0)&&(numRetry<completitionMaxRetry)) {
				rmIndex=0;
				Iterator<Response<DoAbortResponse>> ackIt=abortAck.iterator();
				while(ackIt.hasNext()) {
					Response<DoAbortResponse> ack=ackIt.next();
					try {
						ack.get(completitionSingleTimeout, TimeUnit.MILLISECONDS);
						ackIt.remove();
					} catch (Exception e) {					
						logger.debug("Timeout exceeded: retrasmission needed");
					}
					rmIndex++;
				}
				numRetry++;
			}
		}
		logger.debug("Completition Ack collected!");
		this.transLogger.writeComplete(transId);
		TransactionResponse response=TransactionResponse.Abort;
		if (decision&&trans.checkCompleted()) {
			response=TransactionResponse.Commit;
			logger.debug("Transaction "+transId+" committed");			
		} else {
			logger.debug("Transaction "+transId+" aborted");
		}
		this.repository.completeTransaction(transId);
		return response;
	}

	@Override
	public String openTransaction() {		
		String transId=repository.registerTransaction();
		logger.debug("Transaction started with ID "+transId);
		return transId;
	}
	@Override
	public boolean join(String transId, String rmWsdl) {
		try {
			logger.debug("Join "+rmWsdl+" on "+transId+" transaction");
			URL rmWsdlUrl=new URL(rmWsdl);
			this.transLogger.writeLink(transId, rmWsdl.toString());
			return this.repository.registerJoin(transId, rmWsdlUrl);
		} catch (MalformedURLException mue) {
			return false;
		}
	}
}
